// 版权所有2016 etcd作者
// 
// 根据Apache许可证2.0版（以下简称“许可证”）获得许可；
// 除非遵守许可证，否则不得使用此文件。
// 您可以通过
// 
// www.apache.org/licenses/LICENSE-2.0 
// 
// 除非适用法律要求或书面同意，否则根据许可证分发的软件
// 按“原样”分发，
// 无任何明示或暗示的保证或条件。
// 请参阅许可证，了解管理许可的特定语言和
// 许可证下的限制。

// 包镜像实现etcd镜像操作。
package mirror

import (
	"context"

	"go.etcd.io/etcd/client/v3"
)

const (
	batchLimit = 1000
)

// 同步器与etcd群集的键值状态同步。
type Syncer interface {
	// SyncBase同步键值状态的基态。
	// 键值状态通过返回的chan发送。
	SyncBase(ctx context.Context) (<-chan clientv3.GetResponse, chan error)
	// 同步键值状态的更新。
	// 更新事件通过返回的chan发送。
	SyncUpdates(ctx context.Context) clientv3.WatchChan
}

// 新闻同步器创建同步器。
func NewSyncer(c *clientv3.Client, prefix string, rev int64) Syncer {
	return &syncer{c: c, prefix: prefix, rev: rev}
}

type syncer struct {
	c      *clientv3.Client
	rev    int64
	prefix string
}

func (s *syncer) SyncBase(ctx context.Context) (<-chan clientv3.GetResponse, chan error) {
	respchan := make(chan clientv3.GetResponse, 1024)
	errchan := make(chan error, 1)

	// 如果未指定版本，我们将选择最新版本。
	if s.rev == 0 {
		resp, err := s.c.Get(ctx, "foo")
		if err != nil {
			errchan <- err
			close(respchan)
			close(errchan)
			return respchan, errchan
		}
		s.rev = resp.Header.Revision
	}

	go func() {
		defer close(respchan)
		defer close(errchan)

		var key string

		opts := []clientv3.OpOption{clientv3.WithLimit(batchLimit), clientv3.WithRev(s.rev),
			clientv3.WithSort(clientv3.SortByKey, clientv3.SortAscend)}

		if len(s.prefix) == 0 {
			// 如果len（s.prefix）=0，我们将同步整个键值空间。
			// 然后我们从最小的键（0x00）开始到结束。
			opts = append(opts, clientv3.WithFromKey())
			key = "\x00"
		} else {
			// 如果len（s.前缀）！=0，我们将使用给定前缀同步键值空间。
			// 然后，如果存在前缀，则从前缀到下一个前缀的范围。或者如果下一个前缀不存在，我们将
			// 从前缀到结尾的范围。
			opts = append(opts, clientv3.WithRange(clientv3.GetPrefixRangeEnd(s.prefix)))
			key = s.prefix
		}

		for {
			resp, err := s.c.Get(ctx, key, opts...)
			if err != nil {
				errchan <- err
				return
			}

			respchan <- *resp

			if !resp.More {
				return
			}
			// 移动到下一个键
			key = string(append(resp.Kvs[len(resp.Kvs)-1].Key, 0))
		}
	}()

	return respchan, errchan
}

func (s *syncer) SyncUpdates(ctx context.Context) clientv3.WatchChan {
	if s.rev == 0 {
		panic("unexpected revision = 0. Calling SyncUpdates before SyncBase finishes?")
	}
	return s.c.Watch(ctx, s.prefix, clientv3.WithPrefix(), clientv3.WithRev(s.rev+1))
}
